package com.yunji.common.event;


import com.yunji.common.event.config.RedissonEventConfig;
import com.yunji.common.event.entity.EventData;
import com.yunji.common.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * @author : peter-zhu
 * @date : 2024/11/1 11:51
 * @description : TODO
 **/
@Slf4j
public class RedissonEventBus implements EventBus {

    private static EventBus redissonEventBus;

    private RedissonEventBus() {
    }

    public static EventBus getInstance() {
        if (redissonEventBus == null) {
            redissonEventBus = new RedissonEventBus();
        }
        return redissonEventBus;
    }

    @Override
    public <T> void publishAsync(EventData<T> event) {
        RedissonEventConfig.redissonClient.getBlockingDeque(event.getQueueName())
                .offerAsync(event.getEventData()).whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("RedissonEventBus 异步入队失败:{}", JsonUtil.toJsonString(event.getEventData()), ex);
                    } else if (result) {
                        log.debug("RedissonEventBus 异步入队成功:{}", event.getQueueName());
                    } else {
                        log.warn("RedissonEventBus 异步入队失败：超时");
                    }
                });
    }

    @Override
    public <T> void publish(EventData<T> event) throws Exception {
        boolean offer = RedissonEventConfig.redissonClient.getBlockingDeque(event.getQueueName())
                .offer(event.getEventData(), 3L, TimeUnit.SECONDS);
        if (!offer)
            log.error("异步入队失败:{}", JsonUtil.toJsonString(event.getEventData()));
    }


    // redis 额外的一种 消息订阅方式
    public <T> void publish(String topicName, Object obj) {
        RedissonEventConfig.redissonClient.getTopic(topicName).publishAsync(obj);
    }
}
